package com.gaofans.nrpc.common.protocol;

import com.gaofans.nrpc.common.compress.Codec;
import com.gaofans.nrpc.common.compress.CodecFactory;
import com.gaofans.nrpc.common.serializer.Serializer;
import com.gaofans.nrpc.common.serializer.SerializerFactory;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToMessageCodec;

import java.util.List;

/**
 * 消息转码与解码器
 * @author GaoFans
 */
@ChannelHandler.Sharable
public class MessageCodec extends MessageToMessageCodec<ByteBuf, Message> {

    private final SerializerFactory serializerFactory;
    private final Codec codec;
    private final byte serializerType;
    private final short version;

    public MessageCodec(SerializerFactory serializerFactory, byte serializerType,short version,Codec codec) {
        this.serializerFactory = serializerFactory;
        this.serializerType = serializerType;
        this.version = version;
        this.codec = codec;
    }

    public MessageCodec(Class<? extends ByteBuf> inboundMessageType,
                        Class<? extends Message> outboundMessageType,
                        SerializerFactory serializerFactory,
                        byte serializerType,
                        short version,
                        Codec codec) {
        super(inboundMessageType, outboundMessageType);
        this.serializerFactory = serializerFactory;
        this.serializerType = serializerType;
        this.version = version;
        this.codec = codec;
    }

    @Override
    protected void encode(ChannelHandlerContext ctx, Message message, List<Object> list) throws Exception {
        ByteBuf out = ctx.alloc().buffer();
        Serializer serializer = serializerFactory.getSerializer(serializerType);
        byte[] data = codec.compress(serializer.serialize(message));
        ProtocolHeader header = new ProtocolHeader(version, serializerType, message.getType(), message.getSequenceId(), data.length, codec.getMessageEncoding());
        header.writeBuf(out);
        out.writeBytes(data);
        list.add(out);

    }

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf buf, List<Object> list) throws Exception {
        ProtocolHeader header = new ProtocolHeader(buf);
        byte[] bytes = new byte[header.getLength()];
        buf.readBytes(bytes, 0, header.getLength());
        Serializer serializer = serializerFactory.getSerializer(header.getSerializeType());
        Codec codec = CodecFactory.get(header.getCodecType());
        Message message = serializer.deserialize(codec.decompress(bytes), Message.getMessageClass(header.getMessageType()));
        list.add(message);
    }

}
